

namespace Stee.CAP8.MessageBus.SampleSubscribe
{
    using System;
    using System.Collections.Generic;
    using System.Threading;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Enums;
    using Stee.CAP8.MessageBus.Interface.Interfaces;
    using Stee.CAP8.MessageBus.Interface.Models;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Subscriber;
    using Stee.CAP8.MessageBus.Sample.Model;

    public class Program
    {
        private static readonly Logger logger = LogManager.GetCurrentClassLogger();


        private static readonly string exitCommand = "EXIT";

        private static readonly string topic = "first_topic";
        private static readonly string key = "first_key";
        private static readonly string groupID = "first_group";

        private static readonly string topic2 = "second_topic";
        private static readonly string key2 = "second_key";
        private static readonly string groupID2 = "second_group";

        private static readonly TopicName topicNameString = new TopicName(topic, key, MessageType.Event);
        private static readonly TopicName topicNameObject = new TopicName(topic2, key2, MessageType.Event);

        private static ISubscriber<string> subscriber;
        private static ISubscriber<Person> subscriberObject;
        private static bool isReady = false;
        private static bool isExit = false;

        private static ManualResetEvent quitEvent = new ManualResetEvent(false);


        public static void Main(string[] args)
        {
            TestSubscribe();


            Console.CancelKeyPress += (sender, eArgs) =>
            {
                quitEvent.Set();
                eArgs.Cancel = true;
            };

            //// Kick off asynchronous stuff 
            quitEvent.WaitOne();
        }

        private static void TestSubscribe()
        {
            try
            {
                SubscriberConfiguration subscriberConfig = new SubscriberConfiguration(new Dictionary<string, string>
                {
                    { "BootstrapServers", "127.0.0.1:9092" },
                    { "GroupID", groupID },
                    { "ClientID", "mysubscriber" },
                    { "EnableAutoCommit", "true" },
                    { "AutoCommitIntervalMs", "5000" },
                    { "EnableAutoOffsetStore", "true" },
                });

                subscriber = new SubscriberImpl<string>(topicNameString, subscriberConfig, MessageHandlerString);
                subscriber.Subscribe();

                //SubscriberConfiguration subscriberConfig2 = new SubscriberConfiguration(new Dictionary<string, string>
                //{
                //    { "BootstrapServers", "127.0.0.1:9092" },
                //    { "GroupID", groupID2 },
                //    { "ClientID", "mysubscriber" },
                //    { "EnableAutoCommit", "false" }, // default: true
                //    { "EnableAutoOffsetStore", "true" },
                //});

                //subscriberObject = new SubscriberImpl<Person>(subscriberConfig2, MessageHandlerPerson);
                //subscriberObject.Subscribe(topicNameObject);
            }
            catch (Exception ex)
            {
                logger.Error($"[ERROR] {ex.Message}");
                Console.WriteLine($"[ERROR] {ex.Message}");
                throw;
            }
        }

        /// <summary>
        /// Message handler for string type
        /// </summary>
        /// <param name="topicName"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        private static bool MessageHandlerString(TopicName topicName, string message)
        {
            try
            {
                if (topicName == null)
                {
                    throw new ArgumentNullException(nameof(topicName));
                }

                bool isConsumed = false;

                //// Handle data type, string
                Console.WriteLine($"Handled message <string>: {message.ToString()}");

                isExit = message.Equals(exitCommand, StringComparison.InvariantCultureIgnoreCase);

                isConsumed = true;

                return isConsumed;
            }
            catch (Exception ex)
            {
                logger.Error($"[ERROR] {ex.Message}");
                Console.WriteLine($"[ERROR] {ex.Message}");
                throw;
            }
            finally
            {
                if (isExit)
                {
                    subscriber.Dispose();

                    if (subscriber.IsDisposed)
                    {
                        Console.WriteLine($"Subscriber is disposed.");
                    }
                    else
                    {
                        Console.WriteLine($"[ERROR] Subscriber is not disposed.");
                    }
                }
            }
        }

        /// <summary>
        /// Message handler for Person type
        /// </summary>
        /// <param name="topicName"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        private static bool MessageHandlerPerson(TopicName topicName, Person message)
        {
            try
            {
                if (topicName == null)
                {
                    throw new ArgumentNullException(nameof(topicName));
                }

                bool isConsumed = false;

                //// Handle data type, string
                Console.WriteLine($"Handled message <Person>: {message.ToString()}");
                isConsumed = true;

                return isConsumed;
            }
            catch (Exception ex)
            {
                logger.Error($"[ERROR] {ex.Message}");
                Console.WriteLine($"[ERROR] {ex.Message}");
                throw;
            }
            finally
            {
                if (isExit)
                {
                    subscriberObject.Dispose();

                    if (subscriber.IsDisposed)
                    {
                        Console.WriteLine($"SubscriberObject is disposed.");
                    }
                    else
                    {
                        Console.WriteLine($"[ERROR] SubscriberObject is not disposed.");
                    }
                }                
            }
        }
    }
}
